package com.xnyzc.lhy.order.mq.consumer;

import com.alibaba.druid.support.json.JSONUtils;
import com.alibaba.fastjson.JSON;
import com.xnyzc.lhy.common.component.config.security.RocketConfig;
import com.xnyzc.lhy.common.component.redis.RedisCommon;
import com.xnyzc.lhy.common.constant.MqConstant;
import com.xnyzc.lhy.common.constant.MqConsumerGroupConstant;
import com.xnyzc.lhy.common.util.CheckUtil;
import com.xnyzc.lhy.order.mo.OrderLogMo;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;


/**
 * @author luoqingwei
 **/
@Slf4j
@Component
public class MqOrderLogConsumer {

    private DefaultMQPushConsumer consumer;

    @Autowired
    RocketConfig config;

    @Autowired
    RedisCommon redisCommon;

    @Value("${rocketMq.isStart}")
    private boolean isStart;

    @Autowired
    private MongoTemplate mongoTemplate;

    /**
     * 开启监听
     */
    @PostConstruct
    public void init() {
        try {
            if (!isStart) {
                return;
            }
            consumer = new DefaultMQPushConsumer(MqConsumerGroupConstant.LOG_OPERATE_ORDER_CALL);
            consumer.setNamesrvAddr(config.getNameServerAddr());
            consumer.subscribe(config.getTopic(), MqConstant.LOG_OPERATE_ORDER);
            consumer.setConsumeMessageBatchMaxSize(1);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    MessageExt messageExt = msgs.get(0);
                    try {
                        //发送的消息
                        String body = new String(messageExt.getBody());
                        log.info("mq消费写入订单日志开始,msgId:{}",messageExt.getMsgId());
                        if (CheckUtil.strIsEmpty(body)) {
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        OrderLogMo orderLogVO = JSON.parseObject(body, OrderLogMo.class);
                        log.info("测试日志,mongodb入库日志:{}",JSON.toJSONString(orderLogVO));
                        mongoTemplate.save(orderLogVO);
                    } catch (Throwable e) {
                        log.error("mq消费订单日志业务处理异常:{}", e.getMessage(),e);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            log.info("mq消费写入订单日志完成");
        } catch (Exception e) {
            log.error("mq消费处理订单日志mq异常:{}", e.getMessage(),e);
        }
    }
}
